package com.supreme.simpleserver.consumer;

import com.supreme.common.constant.ConsoleColors;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
public class KafkaConsumer {

    @KafkaListener(topics = {"test", "supreme", "sale_order"}, groupId = "test-json")
    public void onMessage(ConsumerRecord<String, String> record) {
        Thread thread = Thread.currentThread();
        String threadName = thread.getName();
        ConsoleColors.printf(ConsoleColors.RED_BOLD_BRIGHT, "%s START\n", threadName);
        try {
            TimeUnit.SECONDS.sleep(10);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        StringBuilder stringBuilder = new StringBuilder("KafkaConsumer\n");
        stringBuilder.append("threadName = ").append(threadName).append("\n");
        stringBuilder.append("topic = ").append(record.topic()).append("\n");
        stringBuilder.append("partition = ").append(record.partition()).append("\n");
        stringBuilder.append("value = ").append(record.value()).append("\n");
        ConsoleColors.println(ConsoleColors.YELLOW_BOLD_BRIGHT, stringBuilder.toString());

        ConsoleColors.printf(ConsoleColors.RED_BOLD_BRIGHT, "%s END\n", threadName);
    }
}
